package com.paas.form.handler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.codahale.metrics.Counter;
import com.codahale.metrics.MetricRegistry;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.paas.common.core.encryp.AESUtil;
import com.paas.common.model.DataBaseEntity;
import com.paas.common.util.AppUtil;
import com.paas.common.util.BeanUtil;
import com.paas.common.util.StringUtil;
import com.paas.common.web.SystemConst;
import com.paas.db.common.element.conf.etl.EtlConf;
import com.paas.db.common.metrics.MetricCollector;
import com.paas.db.common.metrics.etl.EtlChannelMetric;
import com.paas.db.common.metrics.etl.EtlWriteMetric;
import com.paas.etlx.core.etl.EtlPluginExecutor;
import com.paas.etlx.core.util.UuidUtil;
import com.paas.form.entity.FormTaskEntity;
import com.paas.form.entity.FormTaskEtlLogEntity;
import com.paas.form.entity.FormTaskLogEntity;
import com.paas.form.entity.FormTaskMapperEntity;
import com.paas.form.enums.RunStateEnums;
import com.paas.form.handler.util.FormJobHandlerUtil;
import com.paas.form.service.IFormTaskEtlLogService;
import com.paas.form.service.IFormTaskLogService;
import com.paas.form.service.IFormTaskMapperService;
import com.paas.form.service.IFormTaskService;
import com.paas.plus.tool.form.model.View;
import com.paas.plus.tool.form.server.FormService;
import com.paas.scheduler.exception.SchedulerException;
import com.paas.scheduler.model.JobInstance;
import com.paas.scheduler.spi.JobHandler;
import com.publics.feign.service.IUcDataBaseService;
import org.springframework.core.env.Environment;

import java.util.*;
import java.util.concurrent.TimeUnit;

public class SyncTaskHandler implements JobHandler {

    private IFormTaskService formTaskService;

    private IFormTaskMapperService formTaskMapperService;

    private Environment environment;

    private FormService formService;

    private IUcDataBaseService dataBaseService;

    private IFormTaskLogService formTaskLogService;

    private IFormTaskEtlLogService etlLogService;

    private List<FormTaskMapperEntity> formTaskMapperList;

    private FormTaskEntity formTaskEntity;

    private String appCode;

    private String publicsKey;

    private String conditionForm;

    private String masterTableName;

    private JSONObject readerObj;

    private JobInstance jobInstance;

    private String token;


    @Override
    public String getType() {
        return "SYNC_TASK";
    }

    @Override
    public void init(JobInstance jobInstance) {
        initService();
        initExecuteParams(jobInstance);
    }

    private void initExecuteParams(JobInstance jobInstance) {
        formTaskEntity = formTaskService.getById(jobInstance.getBusId());
        conditionForm = formTaskEntity.getConditionForm();
        String viewId = formTaskEntity.getViewId();
        //获取token
        appCode = environment.getProperty("paas.token.thirdParty.appCode");
        publicsKey = environment.getProperty("paas.token.thirdParty.publicsKey");
        token = FormJobHandlerUtil.getToken(appCode, publicsKey);

        //获取主视图数据库连接信息
        View view = formService.getViewById(viewId);
        String datasourceId = view.getDatasourceId();
        DataBaseEntity dataBaseEntity = dataBaseService.asyncQueryById(datasourceId, token);

        FormTaskMapperEntity formTaskMapperEntity = new FormTaskMapperEntity();
        formTaskMapperEntity.setTaskId(formTaskEntity.getId());
        formTaskMapperList = formTaskMapperService.getList(formTaskMapperEntity);

        masterTableName = view.getLinkTable();
        if (BeanUtil.isEmpty(readerObj)){
            readerObj = new JSONObject();
        }
        readerObj.put("name", dataBaseEntity.getDbType() + "reader");
        //获取主视图reader
        JSONObject readerParameter = new JSONObject();
        readerParameter.put("jdbcUrl", dataBaseEntity.getJdbcUrl());
        readerParameter.put("username", dataBaseEntity.getJdbcUser());
        readerParameter.put("password", dataBaseEntity.getJdbcPassword());
        readerObj.put("parameter", readerParameter);

        this.jobInstance = jobInstance;
    }


    @Override
    public void start() {
        int totalRecordCount = 0;
        Boolean flag = true;
        long startTime = System.currentTimeMillis();
        FormTaskLogEntity formTaskLogEntity = new FormTaskLogEntity();
        formTaskLogEntity.setJobId(jobInstance.getInstId());
        formTaskLogEntity.setTaskId(jobInstance.getBusId());
        MetricCollector metricCollector = null;
        String executeJsonStr = null;
        EtlPluginExecutor etlPluginExecutor = null;
        try {
            executeJsonStr = parseExecuteJson();
            EtlConf etlConf = JSON.parseObject(executeJsonStr, EtlConf.class);
            String etlInstId = UuidUtil.shortUpperUuid();
            String pluginId = etlInstId + "-" + String.format("%04d", 1);
            formTaskLogEntity.setStartTime(new Date());
            etlPluginExecutor = new EtlPluginExecutor(etlConf, null, new MetricRegistry(), jobInstance.getDefId(), jobInstance.getInstId(), pluginId);
            etlPluginExecutor.doStart();
            //任务启动,保存日志为启动
            formTaskLogEntity.setStatus(RunStateEnums.start.getValue());
            formTaskLogService.saveOrUpdate(formTaskLogEntity);
            //当线程数据为执行完成前,进行轮询休眠
            while (true){
                if (!etlPluginExecutor.isTaskFinished()){
                    TimeUnit.NANOSECONDS.sleep(500);
                    //当状态为执行器正在执行中,状态修改为running
                    if (formTaskLogEntity.getStatus().equals(RunStateEnums.start.getValue())){
                        formTaskLogEntity.setStatus(RunStateEnums.running.getValue());
                        formTaskLogService.saveOrUpdate(formTaskLogEntity);
                    }
                }else {
                    break;
                }
            }
        }catch (Exception e){
            flag = false;
            formTaskLogEntity.setError(e.getMessage());
        }finally {
            if (flag){
                formTaskLogEntity.setStatus(RunStateEnums.finish.getValue());
            }else {
                formTaskLogEntity.setStatus(RunStateEnums.error.getValue());
            }
            if (BeanUtil.isNotEmpty(etlPluginExecutor)){
                metricCollector = etlPluginExecutor.getMetricCollector();
                EtlChannelMetric channelMetric = metricCollector.getChannelMetric();
                Counter totalRecords = channelMetric.getTotalRecords();
                totalRecordCount  += totalRecords.getCount();
                saveEtlLog(metricCollector,executeJsonStr);
            }
        }

        long endTime = System.currentTimeMillis();
        formTaskLogEntity.setEndTime(new Date());
        formTaskLogEntity.setRecords(totalRecordCount);
        formTaskLogEntity.setTotalTime(String.valueOf((endTime-startTime)));
        formTaskLogService.saveOrUpdate(formTaskLogEntity);
    }

    private void saveEtlLog(MetricCollector metricCollector, String executeJsonStr) {
        EtlWriteMetric writeMetric = metricCollector.getWriteMetric();
        EtlChannelMetric channelMetric = metricCollector.getChannelMetric();
        Counter totalRecords = channelMetric.getTotalRecords();
        Long successCount =  totalRecords.getCount() - writeMetric.getWriteFailedRecords().getCount();
        FormTaskEtlLogEntity formTaskEtlLogEntity = new FormTaskEtlLogEntity();
        formTaskEtlLogEntity.setTaskId(jobInstance.getBusId());
        formTaskEtlLogEntity.setJobId(jobInstance.getInstId());
        String aesStr = AESUtil.aesEncode(executeJsonStr);
        formTaskEtlLogEntity.setExecuteJson(aesStr);
        formTaskEtlLogEntity.setStatus(SystemConst.NORMAL_STATUS);
        formTaskEtlLogEntity.setFailCount(Integer.valueOf(String.valueOf(writeMetric.getWriteFailedRecords().getCount())));
        formTaskEtlLogEntity.setSuccessCount(Integer.valueOf(successCount.toString()));
        formTaskEtlLogEntity.setDistributionType("table");
        etlLogService.saveOrUpdate(formTaskEtlLogEntity);
    }

    @Override
    public void destroy() {

    }

    @Override
    public JobHandler clone() {
        try {
            return (JobHandler) super.clone();
        } catch (CloneNotSupportedException e) {
            throw new SchedulerException("调度服务克隆异常！", e);
        }
    }

    private String parseExecuteJson() {
        List<String> purposeColumn = new ArrayList<>();
        StringBuilder sourceBuilder = new StringBuilder();
        JSONObject writerObj = new JSONObject();
        for (FormTaskMapperEntity formTaskMapperEntity : formTaskMapperList) {
            if (StringUtil.isNullOrEmpty(sourceBuilder)) {
                sourceBuilder.append(formTaskMapperEntity.getSourceCloumn());
            } else {
                sourceBuilder.append(",").append(formTaskMapperEntity.getSourceCloumn());
            }
            purposeColumn.add(formTaskMapperEntity.getTargetCloumn());
        }

        String readerSql = "select " + sourceBuilder.toString() + " from  " + masterTableName;
        JSONObject readerConnectionObj = new JSONObject();
        JSONObject parameter = readerObj.getJSONObject("parameter");
        if (!StringUtil.isNullOrEmpty(conditionForm)) {
            Map map = JSONObject.parseObject(conditionForm, Map.class);
            String conditionType = map.get("conditionType").toString();
            if ("1".equalsIgnoreCase(conditionType)){
                List list = new ObjectMapper().convertValue(map.get("conditionList"), List.class);
                StringBuilder sqlBuilder = new StringBuilder();
                for (Object obj : list) {
                    Map param = new ObjectMapper().convertValue(obj, Map.class);
                    String code = param.get("code").toString();
                    String value = param.get("value").toString();
                    //条件
                    String type = param.get("type").toString();
                    sqlBuilder.append(" and ").append(code).append(type).append("'"+value+"'");
                }
                parameter.put("where",sqlBuilder.toString());
            }else {
                String conditionDivStr = map.get("conditionDivStr").toString();
                if (conditionDivStr.contains("where")){
                    conditionDivStr.replace("where", " and ");
                }
                parameter.put("where",conditionDivStr);
                readerObj.put("parameter",parameter);
            }
        }
        readerConnectionObj.put("querySql",readerSql);
        parameter.put("connection", Arrays.asList(readerConnectionObj));
        writerObj = parseWriterJson(purposeColumn);
        JSONObject executeJson = new JSONObject();
        executeJson.put("reader",readerObj);
        executeJson.put("writer",writerObj);
        return executeJson.toJSONString();
    }

    private JSONObject parseWriterJson( List<String> purposeColumn) {
        JSONObject writerObj = new JSONObject();
        String databaseId = formTaskEntity.getDatabaseId();
        DataBaseEntity dataBaseEntity = dataBaseService.asyncQueryById(databaseId, token);
        JSONObject writerConnectionObj = new JSONObject();
        writerConnectionObj.put("table",Arrays.asList(formTaskEntity.getTableName()));
        writerObj.put("name",dataBaseEntity.getDbType()+"writer");
        JSONObject writerParameter = new JSONObject();
        String operationType = formTaskEntity.getOperationType();
        writerParameter.put("writeMode",operationType);
        writerParameter.put("connection",Arrays.asList(writerConnectionObj));
        writerParameter.put("jdbcUrl", dataBaseEntity.getJdbcUrl());
        writerParameter.put("username", dataBaseEntity.getJdbcUser());
        writerParameter.put("password", dataBaseEntity.getJdbcPassword());
        writerParameter.put("column",purposeColumn);
        writerObj.put("parameter",writerParameter);
        return writerObj;
    }

    private void initService() {
        if (formTaskService == null) {
            formTaskService = (IFormTaskService) AppUtil.getBean(IFormTaskService.class);
        }

        if (environment == null) {
            environment = (Environment) AppUtil.getBean(Environment.class);
        }

        if (formService == null) {
            formService = (FormService) AppUtil.getBean(FormService.class);
        }

        if (dataBaseService == null) {
            dataBaseService = (IUcDataBaseService) AppUtil.getBean(IUcDataBaseService.class);
        }

        if (formTaskLogService == null){
            formTaskLogService = (IFormTaskLogService) AppUtil.getBean(IFormTaskLogService.class);
        }

        if (etlLogService == null){
            etlLogService = (IFormTaskEtlLogService) AppUtil.getBean(IFormTaskEtlLogService.class);
        }

        if (formTaskMapperService == null){
            formTaskMapperService = (IFormTaskMapperService) AppUtil.getBean(IFormTaskMapperService.class);
        }
    }
}
